package com.taotao.job.schedule;

import com.taotao.job.core.util.CollectionUtil;
import com.taotao.job.dao.JobScheduleMetaMapper;
import com.taotao.job.model.entity.JobScheduleMeta;
import com.taotao.job.schedule.lock.DistributedLock;
import com.taotao.job.schedule.wheeltimer.WheelTimer;
import com.taotao.job.schedule.wheeltimer.WheelTimerTriggerTask;
import com.taotao.job.service.JobScheduleMetaService;
import lombok.extern.slf4j.Slf4j;

import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author fanbeibei
 * @version 1.0
 * @date 2021/2/22 17:34
 */
@Slf4j
public class JobTimeScheduler {
    private JobLogicScheduler jobLogicScheduler;
    private DistributedLock distributedLock;
    private JobScheduleMetaService jobScheduleMetaService;


    /**
     * 时间轮
     */
    private WheelTimer wheelTimer;

    private Thread scheduleThread;
    private boolean scheduleThreadToStop = false;

    public static final long PRE_READ_MS = 5000;    // pre read

    public static final String SCHEDULE_LOCK = "schedule_lock";

    public JobTimeScheduler(JobLogicScheduler jobLogicScheduler, DistributedLock distributedLock,
                            JobScheduleMetaService jobScheduleMetaService) {
        this.jobLogicScheduler = jobLogicScheduler;
        this.distributedLock = distributedLock;
        this.jobScheduleMetaService = jobScheduleMetaService;
    }

    public void start() {
        startScheduleThread();
        startWheelTimer();
    }


    private void startWheelTimer() {
        wheelTimer = new WheelTimer(1, TimeUnit.SECONDS, 60);
    }

    private void stopWheelTimer() {
        wheelTimer.stop();
    }


    private void startScheduleThread() {
        scheduleThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!scheduleThreadToStop) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(PRE_READ_MS - System.currentTimeMillis() % 1000);
                    } catch (InterruptedException e) {
                        if (!scheduleThreadToStop) {
                            log.error(e.getMessage(), e);
                        }
                    }

                    try {
                        boolean locked = distributedLock.lock(SCHEDULE_LOCK);
                        if (!locked) {
                            continue;
                        }

                        log.debug("---------- get lock success -------------");
                        long timestamp = System.currentTimeMillis() + PRE_READ_MS;
                        List<JobScheduleMeta> jsmList = jobScheduleMetaService.getJobForSchedule(timestamp);
                        if (CollectionUtil.isEmpty(jsmList)) {
                            continue;
                        }

                        for (JobScheduleMeta jsm : jsmList) {
                            doSchedule(jsm, timestamp);
                        }

                        //扫描
                    } catch (Exception e) {
                        log.error("", e);
                    } finally {
                        distributedLock.releaseLock(SCHEDULE_LOCK);
                    }
                }
            }
        });
        scheduleThread.setDaemon(true);
        scheduleThread.setName("taotao-job,  JobTimeScheduler#scheduleThread");
        scheduleThread.start();
    }

    private void doSchedule(JobScheduleMeta jsm, long timestamp) {

        try {
            long triggerNextTime = jsm.getNextTriggerTime();
            CronExpression cronExpression = new CronExpression(jsm.getCron());
            while (triggerNextTime < timestamp) {
                if (-1 == triggerNextTime) {
                    //新建的任务
                    triggerNextTime = cronExpression.getNextInvalidTimeAfter(new Date()).getTime();
                    continue;
                }

                final Long triggerNextTime0 = triggerNextTime;
                wheelTimer.addTask(new WheelTimerTriggerTask() {
                    @Override
                    public void run() {
                        //交给逻辑调度器去触发
                        JobSchelueInfo jobSchelueInfo = new JobSchelueInfo.JobSchelueInfoBuilder()
                                .jobCode(jsm.getJobCode()).executeOnce(jsm.getExecuteOnce())
                                .relyFeature(jsm.getRelyFeature()).status(jsm.getStatus())
                                .shouldTriggerTime(triggerNextTime0)
                                .build();
                        jobLogicScheduler.doLogicSchedule(jobSchelueInfo);
                    }
                }, triggerNextTime);
                triggerNextTime = cronExpression.getNextInvalidTimeAfter(new Date(triggerNextTime)).getTime();
            }

            //更新下次触发时间
            jobScheduleMetaService.updateTriggerNextTime(jsm.getId(),triggerNextTime);

        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void stopScheduleThread() {
        scheduleThreadToStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);  // wait
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        if (scheduleThread.getState() != Thread.State.TERMINATED) {
            // interrupt and wait
            scheduleThread.interrupt();
            try {
                scheduleThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }


    public void stop() {
        stopScheduleThread();
        stopWheelTimer();
    }

}
